package com.mysite.seckill.web.config;

import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.NewTopic;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.kafka.core.KafkaAdmin;
import org.springframework.stereotype.Component;

import com.mysite.seckill.commons.common.Constant;

@Component
public class WebApplicationRunner implements ApplicationRunner {

    protected final Logger logger = LoggerFactory.getLogger(this.getClass());

    @Value("${spring.kafka.producer.bootstrap-servers}")
    private String zkServers;

    @Override
    public void run(ApplicationArguments args) throws Exception {

        Map<String, Object> configs = new HashMap<>();
        configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, zkServers);
        KafkaAdmin admin = new KafkaAdmin(configs);
        AdminClient adminClient = AdminClient.create(admin.getConfig());

        Set<String> resultSet = adminClient.listTopics().names().get();
        if (!resultSet.contains(Constant.KAFKA_SECKILL_TOPIC)) {
            NewTopic newTopic = new NewTopic(Constant.KAFKA_SECKILL_TOPIC, 2, (short) 1);
            adminClient.createTopics(Arrays.asList(newTopic));
            adminClient.close(10, TimeUnit.SECONDS);
        }

        logger.info("Web application is started Successfully !");
    }
}
